package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.mapper.ApArticleMapper;
import com.heima.common.constants.ApCommentConstants;
import com.heima.model.article.pojos.ApArticle;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * @author: Zqr
 */
@Slf4j
@Component
public class ArticleCommentListener {
    @Autowired
    private ApArticleMapper apArticleMapper;

    /**
     * 监听评论微服务发送的kafka消息,更新ap_article表中的comment评论数字段
     *
     * @param msg
     */
//    @KafkaListener(topics = ApCommentConstants.AP_COMMENT_SYNC_ARTICLE)
    @KafkaListener(topics = ApCommentConstants.AP_COMMENT_STREAM_OUT_SYNC_ARTICLE)
    public void onMessage(String msg) {
        if (StringUtils.isNotBlank(msg)) {
            log.info("article监听到kafka流式计算后消息[{}]用于更新ap_article表_comment...",msg);
            Map map = JSON.parseObject(msg, Map.class);
            Long articleId = Long.valueOf(map.get("articleId").toString());
            int add = Integer.parseInt(map.get("add").toString());
            ApArticle apArticle = apArticleMapper.selectById(articleId);
            apArticle.setComment(apArticle.getComment() == null ? add : (apArticle.getComment() + add));
            //更新db
            apArticleMapper.updateById(apArticle);
        }
    }




}
